package org.jxkj.app

import java.text.SimpleDateFormat
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.jxkj.data.IDaoProvider
import org.jxkj.util.JzCondition

/**
 * 日考核电量统计
 */
object StatisticDay {


  // agc 日报详情
  def calAgcDay(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    val startTime = time(0)
    val endTime = time(1)
    val startTime2 = time(0).substring(0,10)+" 00:00:00"
    val endTime2 = time(1).substring(0,10)+" 23:59:59"

    //转换时间格式 2021-03-05加
    val sstartTime = startTime.replaceAll( "/", "-" )
    val sendTime = endTime.replaceAll( "/", "-" )

    JzCondition.calData(ss,sc,hTable,Array(startTime2,endTime2),step,3600)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_eo_fm_agc")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    jdbcDF.createOrReplaceTempView("hms_eo_fm_agc")

//  2021-03-05改
    val etldf=ss.sql(s"select date_format(start_time, 'yyyy/MM/dd') start_date,project_name,avg(key_value) key_value,sum(compensate_cost) compensate_cost  from hms_eo_fm_agc where start_time between '$sstartTime' and '$sendTime' group by date_format(start_time, 'yyyy/MM/dd'),project_name ")
    etldf.createOrReplaceTempView("t_etl")
//        ss.sql("select * from t_etl").show(10)

    val resultDf=ss.sql("select start_date,project_name,key_value,b.unit_cnt,case when key_value<1 then b.unit_cnt*30 else 0 end ass_quantity,compensate_cost" +
      ",'GZB' project_id,'10000' org_id,'AGC考核' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time " +
      " from t_etl a left join t_jzgk_cnt b on a.start_date=b.start_time ")
//        resultDf.show()
    val tablename="hms_eo_fm_agc_day"
    val delString=s"delete from $tablename where start_date between '$startTime' and '$endTime' AND sub_org_id='AGC考核'"
    MysqlDml.delete2(delString)

    //写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()

  }


  //  无功日考核电量
  def calVoltageDay(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    import ss.implicits._
    val startTime = time(0)
    val endTime = time(1)

    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_eo_volpassrate")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    jdbcDF.createOrReplaceTempView("hms_eo_volpassrate")

    val etldf=ss.sql("select substr(stat_time,1,10) start_date,project_name" +
      ",max(case when substr(unit_ass_type,1,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,2,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,3,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,4,1)=1 then 1 else 0 end)+" +
      "max(case when substr(unit_ass_type,5,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,6,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,7,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,8,1)=1 then 1 else 0 end)+" +
      "max(case when substr(unit_ass_type,9,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,10,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,11,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,12,1)=1 then 1 else 0 end)+" +
      "max(case when substr(unit_ass_type,13,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,14,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,15,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,16,1)=1 then 1 else 0 end)+" +
      "max(case when substr(unit_ass_type,17,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,18,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,19,1)=1 then 1 else 0 end)+max(case when substr(unit_ass_type,20,1)=1 then 1 else 0 end)+" +
      "max(case when substr(unit_ass_type,21,1)=1 then 1 else 0 end) unit_cnt " +
      s" from hms_eo_volpassrate where substr(stat_time,1,10) between '$startTime' and '$endTime'" +
      " and status<>0 group by substr(stat_time,1,10),project_name ").cache()
    etldf.createOrReplaceTempView("t_ods")
    import ss.implicits._
    val etldf2=ss.sql("select start_date,'全厂' project_name,sum(unit_cnt) unit_cnt from t_ods group by start_date")
    etldf2.union(etldf).createOrReplaceTempView("t_ods2")

    val resultDf=ss.sql("select start_date,project_name,0 key_value,unit_cnt" +
      ",case when project_name='大江' then 2049*0.2*unit_cnt when project_name='二江' then 1090*0.2*unit_cnt when project_name='全厂' then (1090+2049)*0.2*unit_cnt else 0 end ass_quantity" +
      ",0 compensate_cost" +
      ",'GZB' project_id,'10000' org_id,'无功考核' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time " +
      " from t_ods2 a ")
    resultDf.show()
    val tablename="hms_eo_fm_agc_day"
    val delString=s"delete from $tablename where start_date between '$startTime' and '$endTime' AND sub_org_id='无功考核'"
    MysqlDml.delete2(delString)

    //写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()

  }
}
